package com.wang.helloworld.hbase;

import java.io.IOException;
import java.util.Date;
import java.util.Iterator;
import java.util.NavigableMap;
import java.util.UUID;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FamilyFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.QualifierFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;

/**
 * Hello world!
 *
 */
public class HBaseClient 
{
	//private static final int MAX_TABLE_COUNT=10;
	private static final String testTable = "ts:testTable";
	
	private static HConnection hconnection = null;
	private static HBaseAdmin admin = null;
	private static long id=0;
	
	public static synchronized long getUid()
	{
		return ++id;
	}
	
    public static void main( String[] args ) throws IOException
    {
        System.out.println( "Hello World!" );
        
        HBaseClient hbaseClient = new HBaseClient();
    }

	public static byte[][] getSplits(long min_id, long max_id, int chunks) {
		long chunkSize = (max_id - min_id) / chunks;
		byte[][] splits = new byte[chunks + 2][];
		int index = 0;
		for (Long i = min_id; i <= (max_id + chunkSize); i += chunkSize) {
			Long start = i;
			Long end = i + chunkSize - 1;
			splits[index] = (String.valueOf(end)).getBytes();
			System.out.println(start);
			index++;
		}

		return splits;
	}
	
    public HBaseClient() throws IOException
    {
    	Configuration conf = HBaseConfiguration.create();
    	Configuration.addDefaultResource("hbase-site.xml");
    	conf.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
    	conf.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());
    	
    	conf.set("fs.defaultFS", "hdfs://dev.findsing.com:9000/hbase");
    	conf.set("hadoop.home.dir", "/opt/hadoop/hadoop/default");
    	//conf.set("hbase.zookeeper.quorum", "10.232.31.209,10.232.31.210,10.232.31.211");
        //conf.set("hbase.zookeeper.property.clientPort", "3325");

    	// Simple authentication
    	UserGroupInformation ugi = UserGroupInformation.createRemoteUser("hadoop");
    	User user = User.create(ugi);

    	hconnection = HConnectionManager.createConnection(conf, user);
    	try
    	{
	    	admin = new HBaseAdmin(hconnection);

	    	dropTable(testTable);
	    	
	    	listTables();
	    	HTableInterface table = createTable();
	    	
	    	String row1 = "person-" + UUID.randomUUID().toString().substring(0,8);
	    	createJeff(table, row1);
	    	String row2 = "person-" + UUID.randomUUID().toString().substring(0,8);
	    	createSammi(table, row2);
	    	getNames(table);
	    	
	    	/**
	    	 *  By default, tesTablet will maintains 5 history copies for "name" family, so you will 
	    	 *  see the history values of firstName.
	    	 *  
	    	 *  And because "age" only has one copy, you will not see it's history
	    	 */
	    	updateJeff(table, row1);
	    	getNames(table);
	    	getAge(table);

	    	/**
	    	 * Modify testTable to reduce "name" history to one, and upgrade "age" to 5, then update 
	    	 * this record again, you will see "name"'s history has gone, and now "age" has hisptry.
	    	 */
	    	alterTable(table);
	    	updateJeff(table, row1);
	    	getNames(table);
	    	getAge(table);
	    	
	    	/**
	    	 * Delete "Sammi"
	    	 */
	    	deleteColumn(table, row2, "name", "firstName");
	    	getNames(table);

	    	/**
	    	 * Delete Jeff & Wang
	    	 */
	    	deleteFamily(table, row1, "name");
	    	getNames(table);
	    	
	    	/**
	    	 * Delete person Jeff
	    	 */
	    	deleteRow(table, row1);
	    	getNames(table);
	    	
	    	admin.close();
    	}
    	catch(Exception e)
    	{
    		e.printStackTrace();
    	}
    	finally{
	    	hconnection.close();
    	}
    }

	public void listTables() throws IOException
    {
    	HTableDescriptor[] tableDescriptors = hconnection.listTables();
    	for(HTableDescriptor tableDescriptor: tableDescriptors)
    	{
    		TableName tableName = tableDescriptor.getTableName();
    		System.out.println("Table name: " + tableName.getNameAsString());
    	}
    }
    
    public void dropTable(String tableName) throws IOException
    {
    	try {
			admin.disableTable(Bytes.toBytes(tableName));
			admin.deleteTable(Bytes.toBytes(tableName));
		} catch (TableNotFoundException e) {
		}
    	catch(Exception e)
    	{
    		e.printStackTrace();
    	}
    }

    public HTableInterface createTable() throws IOException
    {
    	HTableDescriptor testTableDescriptor = new HTableDescriptor(TableName.valueOf(testTable));
    	testTableDescriptor.addFamily(new HColumnDescriptor(Bytes.toBytes("id")));
    	testTableDescriptor.addFamily(new HColumnDescriptor(Bytes.toBytes("name")).setMaxVersions(5));
    	testTableDescriptor.addFamily(new HColumnDescriptor(Bytes.toBytes("age")));
    	testTableDescriptor.addFamily(new HColumnDescriptor(Bytes.toBytes("gender")));

    	//byte[][] splits = getSplits(1, 100000, 100);
    	admin.createTable(testTableDescriptor/*, splits*/);

    	return admin.getConnection().getTable(Bytes.toBytes(testTable));
    }
    
    /**
     * Alter table to maintain 1 name history copy. And 5 age's copies
     */
    public void alterTable(HTableInterface table) throws IOException
    {
    	HTableDescriptor testTableDescriptor = new HTableDescriptor(table.getTableDescriptor());
    	testTableDescriptor.addFamily(new HColumnDescriptor(Bytes.toBytes("name")).setMaxVersions(1));
    	testTableDescriptor.addFamily(new HColumnDescriptor(Bytes.toBytes("age")).setMaxVersions(5));
    	
    	admin.modifyTable(table.getTableName(), testTableDescriptor);
    }
    
    public static synchronized void insert(HTableInterface table, Put put) throws IOException
    {
    	Get get = new Get(put.getRow());
    	if(table.exists(get))
    	{
    		throw new RuntimeException("Record exists");
    	}
   		//table.incrementColumnValue(Bytes.toBytes(row), Bytes.toBytes("id"), Bytes.toBytes("exists"), 1);
   		table.put(put);
    }
    
    public static synchronized void update(HTableInterface table, Put put) throws IOException
    {
    	Get get = new Get(put.getRow());
    	if(!table.exists(get))
    	{
    		throw new RuntimeException("Record doesn't exist");
    	}
    	
		table.put(put);
    }
    
    public void createJeff(HTableInterface table, String personId) throws IOException
    {
    	/**
    	 *  To add to a row, use Put class.
    	 *  the Bytes class has utility for converting all kinds of java types to byte arrays.
    	 */
    	Put p = new Put(Bytes.toBytes(personId));
    	
    	/**
    	 * To store values to this row, you must specify the column family, 
    	 * column qualifier, and value of the row
    	 */
    	p.add(Bytes.toBytes("id"), Bytes.toBytes("uid"), Bytes.toBytes(getUid()));
    	p.add(Bytes.toBytes("name"), Bytes.toBytes("firstName"), Bytes.toBytes("Jeff"));
    	p.add(Bytes.toBytes("gender"), Bytes.toBytes(""), Bytes.toBytes("Male"));
    	p.add(Bytes.toBytes("age"), Bytes.toBytes(""), Bytes.toBytes(40L));

    	insert(table, p);
    }

    private void updateJeff(HTableInterface table, String personId) throws IOException {
    	Put p = new Put(Bytes.toBytes(personId));
    	p.add(Bytes.toBytes("name"), Bytes.toBytes("firstName"), Bytes.toBytes("Ji"));
    	p.add(Bytes.toBytes("name"), Bytes.toBytes("lastName"), Bytes.toBytes("Wang"));
    	table.incrementColumnValue(Bytes.toBytes(personId), Bytes.toBytes("age"), Bytes.toBytes(""), 1);
    	
    	update(table, p);
	}

    
    public void createSammi(HTableInterface table, String personId) throws IOException
    {
    	Put p = new Put(Bytes.toBytes(personId));

    	p.add(Bytes.toBytes("id"), Bytes.toBytes("uid"), Bytes.toBytes(getUid()));
    	p.add(Bytes.toBytes("name"), Bytes.toBytes("firstName"), Bytes.toBytes("Sammi"));
    	p.add(Bytes.toBytes("gender"), Bytes.toBytes(""), Bytes.toBytes("Fmale"));
    	
    	insert(table, p);
    }

    public void deleteColumn(HTableInterface table, String row, String family, String qualifier) throws IOException
    {
    	Delete delete = new Delete(Bytes.toBytes(row));
    	delete.deleteColumns(Bytes.toBytes(family), Bytes.toBytes(qualifier));
    	
    	table.delete(delete);
    }
    
    public void deleteFamily(HTableInterface table, String row, String family) throws IOException
    {
    	Delete delete = new Delete(Bytes.toBytes(row));
    	delete.deleteFamily(Bytes.toBytes(family));

    	table.delete(delete);
    }
    
    public void deleteRow(HTableInterface table, String row) throws IOException
    {
    	Delete delete = new Delete(Bytes.toBytes(row));
    	
    	table.delete(delete);
    }
    
    /*public void getId(HTableInterface table, String row) throws IOException
    {
    	Get get = new Get(Bytes.toBytes(row));
    	Result result = table.get(get);
    }*/
    
    public void getAge(HTableInterface table) throws IOException
    {
    	// "And" condition
    	FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
    	
    	// Search certain family
    	Filter familyFilter = new FamilyFilter(CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("age")));
    	filterList.addFilter(familyFilter);
    	
    	// Do "Scan"
    	Scan scan = new Scan();
    	scan.setFilter(filterList);
    	scan.setMaxVersions(5);
    	
    	ResultScanner scanner = table.getScanner(scan);
		for (Result row = scanner.next(); row != null; row = scanner.next()) {
			String rowKey = Bytes.toString(row.getRow());
			
			NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> resultMap = row.getMap();
			Iterator<byte[]> rootKeys = resultMap.keySet().iterator();
			while(rootKeys.hasNext())
			{
				byte[] familyByteArray = rootKeys.next();
				String familyName = Bytes.toString(familyByteArray);
				
				NavigableMap<byte[], NavigableMap<Long, byte[]>> columnMap = resultMap.get(familyByteArray);
				
				Iterator<byte[]> columnKeys = columnMap.keySet().iterator();
				while(columnKeys.hasNext())
				{
					byte[] valueKey = columnKeys.next();
					String columnName = Bytes.toString(valueKey);
					
					NavigableMap<Long, byte[]> valueMap = columnMap.get(valueKey);
					Iterator<Long> timestamps = valueMap.keySet().iterator();
					while(timestamps.hasNext())
					{
						Long timestamp = timestamps.next();
						
						Date lastChange = new Date(timestamp);
						Long value = Bytes.toLong(valueMap.get(timestamp));
						System.out.println("Row key: " + rowKey 
								+ "\tFamily: " + familyName
								+ "\tQualifier: " + columnName
								+ " = " + value
								+ " [Timestamp: " + lastChange.toString() + "]");
					}
				}
			}
		}
    }
    
    public void getNames(HTableInterface table) throws IOException
    {
    	// "And" condition
    	FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
    	
    	// Search certain family
    	Filter familyFilter = new FamilyFilter(CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("name")));
    	filterList.addFilter(familyFilter);
    	
    	// Column name qualifies by *Name
    	Filter colFilter = new QualifierFilter(CompareOp.GREATER_OR_EQUAL, new SubstringComparator("Name"));
    	filterList.addFilter(colFilter);

    	// Do "Scan"
    	Scan scan = new Scan();
    	scan.setFilter(filterList);
    	scan.setMaxVersions(5);
    	
    	ResultScanner scanner = table.getScanner(scan);
		for (Result row = scanner.next(); row != null; row = scanner.next()) {
			String rowKey = Bytes.toString(row.getRow());
			
			NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> resultMap = row.getMap();
			Iterator<byte[]> rootKeys = resultMap.keySet().iterator();
			while(rootKeys.hasNext())
			{
				byte[] familyByteArray = rootKeys.next();
				String familyName = Bytes.toString(familyByteArray);
				
				NavigableMap<byte[], NavigableMap<Long, byte[]>> columnMap = resultMap.get(familyByteArray);
				
				Iterator<byte[]> columnKeys = columnMap.keySet().iterator();
				while(columnKeys.hasNext())
				{
					byte[] valueKey = columnKeys.next();
					String columnName = Bytes.toString(valueKey);
					
					NavigableMap<Long, byte[]> valueMap = columnMap.get(valueKey);
					Iterator<Long> timestamps = valueMap.keySet().iterator();
					while(timestamps.hasNext())
					{
						Long timestamp = timestamps.next();
						
						Date lastChange = new Date(timestamp);
						String value = Bytes.toString(valueMap.get(timestamp));
						System.out.println("Row key: " + rowKey 
								+ "\tFamily: " + familyName
								+ "\tQualifier: " + columnName
								+ " = " + value
								+ " [Timestamp: " + lastChange.toString() + "]");
					}
				}
			}
		}
    }
}
